package cn.cup.dmp.utils

import org.apache.spark.sql.Row

object RptKpiHandler {
  /**
    * 根据row得到kpi指标
    * @param row
    * @return List（原始请求书，有效请求数，满足广告请求数，参与竞价数，成功竞价数，每次消费金额，每次成本，展示了，点击量）
    */
  def rptKpi(row: Row): List[Double] = {
    //提取关键字段：ADPLATFORMPROVIDERID	REQUESTMODE	PROCESSNODE	ISEFFECTIVE	ISBILLING	ISBID	ISWIN	ADORDERID	ADCREATIVEID
    val requestmode = row.getAs[Int]("requestmode")
    val processnode = row.getAs[Int]("processnode")
    val iseffective = row.getAs[Int]("iseffective")
    val isbilling = row.getAs[Int]("isbilling")
    val isbid = row.getAs[Int]("isbid")
    val iswin = row.getAs[Int]("iswin")
    val adorderid = row.getAs[Int]("adorderid")
    val winprice = row.getAs[Double]("winprice")
    //每次消费
    val adpayment = row.getAs[Double]("adpayment") //每次成本
    /*原始请求数,有效请求数量,满足广告请求条件的请求数量*/
    val res = if (requestmode == 1 && processnode == 1) {
      List[Double](1, 0, 0)
    } else if (requestmode == 1 && processnode == 2) {
      List[Double](1, 1, 0)
    } else if (requestmode == 1 && processnode == 3) {
      List[Double](1, 1, 1)
    } else {
      List[Double](0, 0, 0)
    }
    // 参与竞价的次数,成功竞价的次数,每次消费,每次成本
    val res2 = if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) {
      List[Double](1, 0, 0, 0)
    } else if (iseffective == 1 && isbilling == 1 && iswin == 1) {
      List[Double](0, 1, winprice / 1000, adpayment / 1000)
    } else {
      List[Double](0, 0, 0, 0)
    }
    //展示量 点击量
    val res3 = if (requestmode == 2 && iseffective == 1) {
      List[Double](1, 0)
    } else if (requestmode == 3 && iseffective == 1) {
      List[Double](0, 1)
    } else {
      List[Double](0, 0)
    }
    res ++ res2 ++ res3
  }
}
